package edu.gatech.doi.spamframework.mapping;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import weka.core.Attribute;
import weka.core.FastVector;
import weka.core.Instance;
import weka.core.Instances;
import weka.core.converters.ArffSaver;



import edu.gatech.doi.spamframework.core.MappingXMLFiles;
import edu.gatech.doi.spamframework.core.ModelXMLFiles;
import edu.gatech.doi.spamframework.model.Message;
import edu.gatech.doi.spamframework.model.ModelAttribute;

import edu.gatech.doi.spamframework.model.XMLOperator;
import edu.gatech.doi.spamframework.parsers.HtmlParser;


public class TwitterMapper implements Mapper {
	private String modelsdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"models"+System.getProperties().getProperty("file.separator");
	
	private String mappingsdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"mappings"+System.getProperties().getProperty("file.separator");
	private String propertiesdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"properties"+System.getProperties().getProperty("file.separator");
	private String instancesdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"instances"+System.getProperties().getProperty("file.separator");
	private String datasetsdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"datasets"+System.getProperties().getProperty("file.separator")+"Twitter"+System.getProperties().getProperty("file.separator");
	
	private String personmodelfilename="personmodel.xml";
	
	private String messagemodelfilename="messagemodel.xml";
	
	private String personmappingfilename="twitterpersonmapping.xml";
	private String messagemappingfilename="twittermessagemapping.xml";
	private String webpagemodelfilename;
	private String webpagemappingfilename;
	
	private HashMap personmodelhm;
	
	private HashMap personmappinghm;
	
	private HashMap messagemodelhm;
	
	private HashMap messagemappinghm;
	
	private HashMap webpagemodelhm;
	
	private HashMap webpagemappinghm;
	
	private String twitterpersonarfffilename;
	
	private String twittermessagearfffilename;
	
	private String twitterwebpagearfffilename;
	
	private String URL;
	
	private String username;
	
	private String password;

	//0: append; 1: replace
	private int isappended=0;
	
	private HashMap personspamhm=null;
	
	private HashMap messagespamhm=null;
	
	public void init()
	{	
		  //Get the jvm heap size.  

		
		long heapSize = Runtime.getRuntime().totalMemory();  
		
		//Print the jvm heap size.  

	    System.out.println("Heap Size = " + heapSize);  

		
		// load property file for myspace
		
		String filepath0=this.propertiesdirectory+"twitter.properties";
//		System.out.println(filepath0);
		PropertiesFileOperator pfo=new PropertiesFileOperator(filepath0);
		pfo.load();
		HashMap mhm=ModelXMLFiles.getMap();
		HashMap mappinghm=MappingXMLFiles.getMap("TWITTER");
		this.messagemodelfilename=(String) mhm.get("MESSAGE");
		this.personmappingfilename=(String) mappinghm.get("TWITTER.PERSON");
		this.personmodelfilename=(String) mhm.get("PERSON");
		this.webpagemodelfilename=(String) mhm.get("WEBPAGE");
		this.webpagemappingfilename=(String) mappinghm.get("TWITTER.WEBPAGE");
		this.twitterwebpagearfffilename=pfo.getProperty("WEBPAGEARFF");
		this.twittermessagearfffilename=pfo.getProperty("MESSAGEARFF");
		this.twitterpersonarfffilename=pfo.getProperty("PERSONARFF");
		this.isappended=Integer.parseInt(pfo.getProperty("ISAPPENDED"));
		this.URL=pfo.getProperty("URL");
		this.username=pfo.getProperty("USERNAME");
		this.password=pfo.getProperty("PASSWORD");
		
		if(pfo.getProperty("INSTANCESDIR")!=null)
		{
			this.instancesdirectory=pfo.getProperty("INSTANCESDIR");
		}
		if(pfo.getProperty("DATASETSDIR")!=null)
		{
			this.datasetsdirectory=pfo.getProperty("DATASETSDIR");
		}
		
//		System.out.println(this.twitterpersonarfffilename);
//		this.twitterpersongarfffilename="twitterperson.arff";
//		
//		this.twittermessagearfffilename="twittermessage.arff";
		
//		System.out.println(this.messagemodelfilename+" "+this.personmappingfilename+" "+this.personmodelfilename);
		
		// load models and mappings
		XMLOperator xmlo=new XMLOperator();
		//person models
		String filepath1=this.modelsdirectory+this.personmodelfilename;
		this.personmodelhm=new HashMap();
//		System.out.println("filepath:"+filepath);
		this.personmodelhm=xmlo.getModelAttrsFromXML(filepath1);
//		if(this.personmodelhm==null)
//		{
//			System.out.println("personmappinghm is null");
//		}
		String filepath2=this.mappingsdirectory+this.personmappingfilename;
		this.personmappinghm=new HashMap();
		this.personmappinghm=xmlo.getMappingAttrsFromXML(filepath2);
		
		
		//message models
		String filepath3=this.modelsdirectory+this.messagemodelfilename;
		this.messagemodelhm=new HashMap();
//		System.out.println("filepath:"+filepath);
		this.messagemodelhm=xmlo.getModelAttrsFromXML(filepath3);
//		if(this.personmodelhm==null)
//		{
//			System.out.println("personmappinghm is null");
//		}
		String filepath4=this.mappingsdirectory+this.messagemappingfilename;
		this.messagemappinghm=new HashMap();
		this.messagemappinghm=xmlo.getMappingAttrsFromXML(filepath4);
		
		//webpage models
		String filepath5=this.modelsdirectory+this.webpagemodelfilename;
		this.webpagemodelhm=new HashMap();
//		System.out.println("filepath:"+filepath);
		this.webpagemodelhm=xmlo.getModelAttrsFromXML(filepath5);
//		if(this.personmodelhm==null)
//		{
//			System.out.println("personmappinghm is null");
//		}
		String filepath6=this.mappingsdirectory+this.webpagemappingfilename;
		this.webpagemappinghm=new HashMap();
		System.out.println(filepath6);
		this.webpagemappinghm=xmlo.getMappingAttrsFromXML(filepath6);
		
		
		
		
		String personspamfilename=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"twitterspamsenderidL1.txt";
		String messagespamfilename=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"suspended_status_ids";
		File personspamfile=new File(personspamfilename);
		File messagespamfile=new File(messagespamfilename);
		this.personspamhm=new HashMap();
		this.messagespamhm=new HashMap();
		try {
			FileReader frs=new FileReader(personspamfile);
			BufferedReader br=new BufferedReader(frs);
			String line=null;
			while((line=br.readLine())!=null)
			{
				this.personspamhm.put(line.trim(), "1");
			}
			br.close();
			frs.close();
			
			FileReader frs2=new FileReader(messagespamfile);
			BufferedReader br2=new BufferedReader(frs2);
			
			while((line=br2.readLine())!=null)
			{
				this.messagespamhm.put(line.trim(), "1");
			}
			br2.close();
			frs2.close();
			
			
			
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		

		
	}
	
	public Instances parseWebPages() {
		// TODO Auto-generated method stub
		Instances       data = null;
		FastVector      atts;
		/**
		 * attributes list
		 */
		
		ArrayList attributelist=new ArrayList();
		/**
		 * id name location description  url
		 */
//		attributelist.add("ID");
//		attributelist.add("NAME");
//		attributelist.add("LOCATION");
//		attributelist.add("DESCRIPTION");
//		attributelist.add("URL");
//		
		Iterator attrit=this.webpagemappinghm.keySet().iterator();
		HashMap mappingattrhm=new HashMap();
		while(attrit.hasNext())
		{
			String tempattrstr=(String) attrit.next();
			attributelist.add(tempattrstr);
			MappingAttribute mappinga=(MappingAttribute) this.webpagemappinghm.get(tempattrstr);
			mappingattrhm.put(mappinga.getModelAttributeName(), mappinga);
		}
		
		/**
		 * connect to db :Mysql
		 */
		
		Connection conn = null;

		try {
		System.out.println("before connecting...");
		conn =DriverManager.getConnection(this.URL+
		"?user="+this.username+"&password="+this.password);
		System.out.println("connect successfully");
		// Do something with the Connection
		Statement stmt = null;
		ResultSet rs = null;
		try {
		stmt = (Statement) conn.createStatement();
		//rs = stmt.executeQuery("SELECT foo FROM bar");
		// or alternatively, if you don't know ahead of time that
		// the query will be a SELECT...
		
		 stmt.execute("set search_path to twitware,public;");
		 System.out.println("set search_path to twitware,public;");
		 stmt.setFetchSize(100); 
		 
			// 1. set up attributes
			
		    atts = new FastVector();
		    /**
			 * id name location description  url
			 * 
			 */
		   Set personmodelkeyset= this.webpagemodelhm.keySet();
		   List list=new ArrayList(personmodelkeyset);
		   Collections.sort(list);
		    Iterator it=list.iterator();
		    while(it.hasNext())
		    {
		    	String tempstr=(String)it.next();
//		    	System.out.println("tempstr1="+tempstr);
		    	ModelAttribute attribute=(ModelAttribute)this.webpagemodelhm.get(tempstr.trim());
		    	String attributename=attribute.getAttributeName();
//		    	System.out.println("attributename:"+attributename);
		    	int typeid=attribute.getAttributeTypeId();
		    	
		    	switch(typeid)
		    	{
		    	case 1:
		    		//-numeric
		    		
		    		atts.addElement(new Attribute(attributename));
		    		break;
		    	case 2:
		    		//-nominal
		    		FastVector attVals;
		    	    attVals = new FastVector();
		    	    ArrayList al=attribute.getNomialValues();
		    	    Iterator ital=al.iterator();
		    	    while(ital.hasNext())
		    	    {
		    	    	String alstr=(String) ital.next();
		    	    	attVals.addElement(alstr);
		    	    }
		    	    atts.addElement(new Attribute(attributename, attVals));
		    	    break;
		    		
		    	case 3:
		    		//-string
		    		atts.addElement(new Attribute(attributename, (FastVector) null));
		    		break;
		    	case 4:
		    		//-date
		    	    atts.addElement(new Attribute(attributename, (FastVector) null));
		    	    break;
		    	}
		    }
		 // - string
		    // instance_id and label
		    Attribute instanceid=new Attribute("INSTANCE_ID", (FastVector) null);
		    Attribute label=new Attribute("LABEL", (FastVector) null);
		    atts.addElement(instanceid);
		    atts.addElement(label);
		    
		    
		 // 2. create Instances object
		    data = new Instances("TwitterWebPages", atts, 0);
		    
		    Instances tempdata=data;
		    HashMap hm=this.webpageMapping(attributelist);
		    
		   
		    FileWriter writer = null;
		    BufferedWriter bw=null;
	    	 
	    	 int offsetnum = 0;
			 int totalnum=0;
			 int limit=5000;
			 HashMap inhm=new HashMap();
	
			   
			 //create file
			    String arfffilename=this.instancesdirectory+offsetnum+"_"+this.twitterwebpagearfffilename;   
			    File arfffile=new File(arfffilename);
			    if(!arfffile.exists())
			    {
			    	try {
			    		arfffile.createNewFile();
					 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterwebpagearfffilename);
		 
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    else
			    {
			    	try {
				    	if(this.isappended==1)
			    		{
							arfffile.delete();
							arfffile.createNewFile();
						 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterpersonarfffilename);
			    		}
			    	
		    		} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    
			// if()
			 stmt.execute("SELECT file_path,status_id,url_id FROM web_content where file_path not like '%_redirect%';");
			
			// System.out.println("SELECT id,name,location,description,url FROM user_info limit 5000 offset "+offsetnum+";");
			 System.out.println("start query "); 
			rs = stmt.getResultSet();
			System.out.println("done query "); 
		    

		
		// Now do something with the ResultSet ....
		 // convert resultset to instances
			System.out.println("converting resultset to instances");
			writer =new FileWriter(arfffilename, true);
	    	
		    bw=new BufferedWriter(writer);
			if(rs!=null)
			{
		
			 // 3. fill with data
	
			    
			    
			    int num=0;
			    HashMap urlhm=new HashMap();
			    while(rs.next())
			    {
				    num++;
			    	System.out.println("num0="+num);
			    	/**
					 * id name location description  url
					 */
				  String file_path=rs.getString("file_path").trim();
				  String status_id=rs.getString("status_id").trim();
				  String url_id=rs.getString("url_id").trim();
				  
//				  if(urlhm.containsKey(url_id))
//				  {
//					  continue;
//				  }
//				  else
//				  {
//					  Integer value=Integer.valueOf("1");
//					  urlhm.put(url_id, value);
//				  }
				  
				  int webcontentindex=file_path.indexOf("web_content");
				  
				  String HtmlFile=this.datasetsdirectory+file_path.substring(webcontentindex);
				    
				  //HtmlParser
					 HtmlParser hp=new HtmlParser();
				    
						
				if(hp.parseHTMLDoc(HtmlFile))
				{
					
						 // 3. fill with data
					
						 
						
//						    	System.out.println("num0="+num);
						    	/**
								 * id name location description  url
								 */
						    	
			    	Iterator it1=attributelist.iterator();
					while(it1.hasNext())
					{
						String attstr=(String)it1.next();
						String rows = null;
						if(attstr.equalsIgnoreCase("BODY"))
						{
							rows=hp.getBodyText();
						}
						else
						{
//										System.out.println(attstr);
							rows= (String) hp.getHeaderMap().get(attstr);
						}
//									System.out.println(rows);
						ModelAttribute attribute=(ModelAttribute)hm.get(attstr);
						String attributename=attribute.getAttributeName();
//									System.out.println(attributename.trim());
//									System.out.println(rows);
						inhm.put(attributename.trim(), rows);
					}
				    Instance newInst = new Instance(data.numAttributes());
				    
				
						//String key=(String)inhm.get();
						if(this.messagespamhm.containsKey(status_id))
						{
							newInst.setValue(data.attribute(data.numAttributes()-1), "spam");
							
						}
						else
						{
							newInst.setValue(data.attribute(data.numAttributes()-1), "ham");
						}
					
				    
					for(int i=0;i<data.numAttributes();i++)
					{
						
		//				System.out.println("i="+data.numAttributes());
						if(inhm.containsKey(data.attribute(i).name()))
						{
							if(inhm.get(data.attribute(i).name())==null)
								continue;
							
							
							
							
							
						 //-numeric
						   if(data.attribute(i).isNumeric())
						   {
							   newInst.setValue(data.attribute(i),Double.valueOf((String) inhm.get(data.attribute(i).name())));
							   
						   }
						   //-nominal
						   if(data.attribute(i).isNominal())
						   {
							   MappingAttribute mapa=(MappingAttribute) mappingattrhm.get(data.attribute(i).name());
							   ModelAttribute modela=(ModelAttribute) hm.get(mapa.getAttributeName());
							   String tempstr=(String) inhm.get(data.attribute(i).name());
							   ArrayList modelal=modela.getNomialValues();
							   ArrayList mappingal=mapa.getNomialValues();
							   HashMap mappingnominal=new HashMap();
							   for(int n=0;n<modelal.size();n++)
							   {
//								   System.out.println("mapping:"+mappingal.get(n)+ modelal.get(n));
								   mappingnominal.put(mappingal.get(n), modelal.get(n));
							   }
							   
							   newInst.setValue(data.attribute(i),(String)mappingnominal.get(tempstr));
						   }
						   //-string
						   if(data.attribute(i).isString())
						   {
		//					   System.out.println("--"+data.attribute(i).name());
							   String tempstr=(String)inhm.get(data.attribute(i).name());
	//						   System.out.println(tempstr);
							   
							   newInst.setValue(data.attribute(i),tempstr);
								
		//					   vals[i] = data.attribute(i).addStringValue("This is a string!");
						   }
					   
						   //-date
						   if(data.attribute(i).isDate())
						   {
							   newInst.setValue(data.attribute(i),(String) inhm.get(data.attribute(i).name()) );
								
		//					   try {
		//						vals[i] = data.attribute(i).parseDate("2001-11-09");
		//					} catch (ParseException e) {
		//						// TODO Auto-generated catch block
		//						e.printStackTrace();
		//					}
						   }
						}
					}
					
					int id=totalnum+num;
					String instanceidstr=status_id;
					
					newInst.setValue(data.attribute(data.numAttributes()-2), instanceidstr);
					
					
				    // add
				    data.add(newInst);
				    newInst=null;
					 //append new instance to the file
			    	
			    	
				    try {
				    	
				    	  // You are now at the end of the file,
				    	  // and can start writing new data out, e.g.
				    	 bw.newLine(); 
				    	 for(int i=0;i<data.numInstances();i++)
				    	 {
				    		 bw.write(data.instance(i).toString());
				    		 bw.newLine();
				    		 bw.flush();
				    	 }
				    	
				    	} catch (IOException ex ) {
				    	  ex.printStackTrace();
				    	}
				    	// delete instances in data
				    data.delete();
				
				    
				    
				}
				
			    }
			    System.out.println("num="+num);
			    totalnum=totalnum+num;
			    System.out.println("totalnum="+totalnum);
			   
			 // 4. output data
	//		    System.out.println(data);
			    try {
					Thread.sleep(10000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			    
			}
			 bw.close();
	    	 writer.close();
	    	 bw=null;
	    	 writer=null;
		    
//		    Runtime.getRuntime().gc();
			offsetnum=offsetnum+limit;
		
		}
		catch (SQLException ex){
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally {
		// it is a good idea to release
		// resources in a finally{} block
		// in reverse-order of their creation
		// if they are no-longer needed
			
			
			
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException sqlEx) { } // ignore
					rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException sqlEx) { } // ignore
					stmt = null;
			}
		}
		
		} catch (SQLException ex) {
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		}
		
		return data;
	}
	
	
	public HashSet<String> loadFile2HashSet(String filename)
	{
		HashSet<String> temphs=new HashSet<String>();
		try {
			FileReader frs=new FileReader(filename);
			BufferedReader br=new BufferedReader(frs);
			String line=null;
			while((line=br.readLine())!=null)
			{
				temphs.add(line.trim());
			}
			br.close();
			frs.close();
			
			
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return temphs;
	}
	
	
	public Instances parseWebPagesByUrlID(String filename) {
		// TODO Auto-generated method stub
		Instances       data = null;
		FastVector      atts;
		/**
		 * attributes list
		 */
		HashSet<String> urlhs=this.loadFile2HashSet(filename);
		ArrayList attributelist=new ArrayList();
		/**
		 * id name location description  url
		 */
//		attributelist.add("ID");
//		attributelist.add("NAME");
//		attributelist.add("LOCATION");
//		attributelist.add("DESCRIPTION");
//		attributelist.add("URL");
//		
		Iterator attrit=this.webpagemappinghm.keySet().iterator();
		HashMap mappingattrhm=new HashMap();
		while(attrit.hasNext())
		{
			String tempattrstr=(String) attrit.next();
			attributelist.add(tempattrstr);
			MappingAttribute mappinga=(MappingAttribute) this.webpagemappinghm.get(tempattrstr);
			mappingattrhm.put(mappinga.getModelAttributeName(), mappinga);
		}
		
		/**
		 * connect to db :Mysql
		 */
		
		Connection conn = null;

		try {
		System.out.println("before connecting...");
		conn =DriverManager.getConnection(this.URL+
		"?user="+this.username+"&password="+this.password);
		System.out.println("connect successfully");
		// Do something with the Connection
		Statement stmt = null;
		ResultSet rs = null;
		try {
		stmt = (Statement) conn.createStatement();
		//rs = stmt.executeQuery("SELECT foo FROM bar");
		// or alternatively, if you don't know ahead of time that
		// the query will be a SELECT...
		
		 stmt.execute("set search_path to twitware,public;");
		 System.out.println("set search_path to twitware,public;");
		 stmt.setFetchSize(100); 
		 
			// 1. set up attributes
			
		    atts = new FastVector();
		    /**
			 * id name location description  url
			 * 
			 */
		   Set personmodelkeyset= this.webpagemodelhm.keySet();
		   List list=new ArrayList(personmodelkeyset);
		   Collections.sort(list);
		    Iterator it=list.iterator();
		    while(it.hasNext())
		    {
		    	String tempstr=(String)it.next();
//		    	System.out.println("tempstr1="+tempstr);
		    	ModelAttribute attribute=(ModelAttribute)this.webpagemodelhm.get(tempstr.trim());
		    	String attributename=attribute.getAttributeName();
//		    	System.out.println("attributename:"+attributename);
		    	int typeid=attribute.getAttributeTypeId();
		    	
		    	switch(typeid)
		    	{
		    	case 1:
		    		//-numeric
		    		
		    		atts.addElement(new Attribute(attributename));
		    		break;
		    	case 2:
		    		//-nominal
		    		FastVector attVals;
		    	    attVals = new FastVector();
		    	    ArrayList al=attribute.getNomialValues();
		    	    Iterator ital=al.iterator();
		    	    while(ital.hasNext())
		    	    {
		    	    	String alstr=(String) ital.next();
		    	    	attVals.addElement(alstr);
		    	    }
		    	    atts.addElement(new Attribute(attributename, attVals));
		    	    break;
		    		
		    	case 3:
		    		//-string
		    		atts.addElement(new Attribute(attributename, (FastVector) null));
		    		break;
		    	case 4:
		    		//-date
		    	    atts.addElement(new Attribute(attributename, (FastVector) null));
		    	    break;
		    	}
		    }
		 // - string
		    // instance_id and label
		    Attribute instanceid=new Attribute("INSTANCE_ID", (FastVector) null);
		    Attribute label=new Attribute("LABEL", (FastVector) null);
		    atts.addElement(instanceid);
		    atts.addElement(label);
		    
		    
		 // 2. create Instances object
		    data = new Instances("TwitterWebPages", atts, 0);
		    
		    Instances tempdata=data;
		    HashMap hm=this.webpageMapping(attributelist);
		    
		   
		    FileWriter writer = null;
		    BufferedWriter bw=null;
	    	 
	    	 int offsetnum = 0;
			 int totalnum=0;
			 int limit=5000;
			 HashMap inhm=new HashMap();
	
			   
			 //create file
			    String arfffilename=this.instancesdirectory+offsetnum+"_"+this.twitterwebpagearfffilename;   
			    File arfffile=new File(arfffilename);
			    if(!arfffile.exists())
			    {
			    	try {
			    		arfffile.createNewFile();
					 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterwebpagearfffilename);
		 
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    else
			    {
			    	try {
				    	if(this.isappended==1)
			    		{
							arfffile.delete();
							arfffile.createNewFile();
						 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterpersonarfffilename);
			    		}
			    	
		    		} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    
			// if()


			    



			    // Now do something with the ResultSet ....
			    // convert resultset to instances
			    System.out.println("converting resultset to instances");
			    writer =new FileWriter(arfffilename, true);

			    bw=new BufferedWriter(writer);
			    int num=0;
			 for(String urlid:urlhs)
			 {
				
			    stmt.execute("SELECT file_path,status_id,url_id FROM web_content where url_id="+urlid+" and file_path not like '%_redirect%';");

			    // System.out.println("SELECT id,name,location,description,url FROM user_info limit 5000 offset "+offsetnum+";");
			    System.out.println("start query "); 
			    rs = stmt.getResultSet();
			    System.out.println("done query "); 
			    
			    if(rs!=null)
			    {

			    	// 3. fill with data



			    	
			    	HashMap urlhm=new HashMap();
			    	if(rs.next())
			    	{

			    		/**
			    		 * id name location description  url
			    		 */
			    		String file_path=rs.getString("file_path").trim();
			    		String status_id=rs.getString("status_id").trim();
			    		String url_id=rs.getString("url_id").trim();



			    		System.out.println("url_id:"+url_id+" is not in the list.");
			    		num++;
			    		System.out.println("num0="+num);

			    		int webcontentindex=file_path.indexOf("web_content");

			    		String HtmlFile=this.datasetsdirectory+file_path.substring(webcontentindex);

			    		//HtmlParser
			    		HtmlParser hp=new HtmlParser();


			    		if(hp.parseHTMLDoc(HtmlFile))
			    		{

			    			// 3. fill with data



			    			//						    	System.out.println("num0="+num);
			    			/**
			    			 * id name location description  url
			    			 */

			    			Iterator it1=attributelist.iterator();
			    			while(it1.hasNext())
			    			{
			    				String attstr=(String)it1.next();
			    				String rows = null;
			    				if(attstr.equalsIgnoreCase("BODY"))
			    				{
			    					rows=hp.getBodyText();
			    				}
			    				else
			    				{
			    					//										System.out.println(attstr);
			    					rows= (String) hp.getHeaderMap().get(attstr);
			    				}
			    				//									System.out.println(rows);
			    				ModelAttribute attribute=(ModelAttribute)hm.get(attstr);
			    				String attributename=attribute.getAttributeName();
			    				//									System.out.println(attributename.trim());
			    				//									System.out.println(rows);
			    				inhm.put(attributename.trim(), rows);
			    			}
			    			Instance newInst = new Instance(data.numAttributes());


			    			//String key=(String)inhm.get();
			    			if(this.messagespamhm.containsKey(status_id))
			    			{
			    				newInst.setValue(data.attribute(data.numAttributes()-1), "spam");

			    			}
			    			else
			    			{
			    				newInst.setValue(data.attribute(data.numAttributes()-1), "ham");
			    			}


			    			for(int i=0;i<data.numAttributes();i++)
			    			{

			    				//				System.out.println("i="+data.numAttributes());
			    				if(inhm.containsKey(data.attribute(i).name()))
			    				{
			    					if(inhm.get(data.attribute(i).name())==null)
			    						continue;





			    					//-numeric
			    					if(data.attribute(i).isNumeric())
			    					{
			    						newInst.setValue(data.attribute(i),Double.valueOf((String) inhm.get(data.attribute(i).name())));

			    					}
			    					//-nominal
			    					if(data.attribute(i).isNominal())
			    					{
			    						MappingAttribute mapa=(MappingAttribute) mappingattrhm.get(data.attribute(i).name());
			    						ModelAttribute modela=(ModelAttribute) hm.get(mapa.getAttributeName());
			    						String tempstr=(String) inhm.get(data.attribute(i).name());
			    						ArrayList modelal=modela.getNomialValues();
			    						ArrayList mappingal=mapa.getNomialValues();
			    						HashMap mappingnominal=new HashMap();
			    						for(int n=0;n<modelal.size();n++)
			    						{
			    							//								   System.out.println("mapping:"+mappingal.get(n)+ modelal.get(n));
			    							mappingnominal.put(mappingal.get(n), modelal.get(n));
			    						}

			    						newInst.setValue(data.attribute(i),(String)mappingnominal.get(tempstr));
			    					}
			    					//-string
			    					if(data.attribute(i).isString())
			    					{
			    						//					   System.out.println("--"+data.attribute(i).name());
			    						String tempstr=(String)inhm.get(data.attribute(i).name());
			    						//						   System.out.println(tempstr);

			    						newInst.setValue(data.attribute(i),tempstr);

			    						//					   vals[i] = data.attribute(i).addStringValue("This is a string!");
			    					}

			    					//-date
			    					if(data.attribute(i).isDate())
			    					{
			    						newInst.setValue(data.attribute(i),(String) inhm.get(data.attribute(i).name()) );

			    						//					   try {
			    						//						vals[i] = data.attribute(i).parseDate("2001-11-09");
			    						//					} catch (ParseException e) {
			    						//						// TODO Auto-generated catch block
			    						//						e.printStackTrace();
			    						//					}
			    					}
			    				}
			    			}

			    			int id=totalnum+num;
			    			String instanceidstr=url_id;

			    			newInst.setValue(data.attribute(data.numAttributes()-2), instanceidstr);


			    			// add
			    			data.add(newInst);
			    			newInst=null;
			    			//append new instance to the file


			    			try {

			    				// You are now at the end of the file,
			    				// and can start writing new data out, e.g.
			    				bw.newLine(); 
			    				for(int i=0;i<data.numInstances();i++)
			    				{
			    					bw.write(data.instance(i).toString());
			    					bw.newLine();
			    					bw.flush();
			    				}

			    			} catch (IOException ex ) {
			    				ex.printStackTrace();
			    			}
			    			// delete instances in data
			    			data.delete();



			    		}

			    	}
			    }
			    System.out.println("num="+num);
			    totalnum=totalnum+num;
			    System.out.println("totalnum="+totalnum);
			   
			 // 4. output data
	//		    System.out.println(data);
//			    try {
//					Thread.sleep(10000);
//				} catch (InterruptedException e) {
//					// TODO Auto-generated catch block
//					e.printStackTrace();
//				}
			    
			}
			 bw.close();
	    	 writer.close();
	    	 bw=null;
	    	 writer=null;
		    
//		    Runtime.getRuntime().gc();
			offsetnum=offsetnum+limit;
		
		}
		catch (SQLException ex){
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally {
		// it is a good idea to release
		// resources in a finally{} block
		// in reverse-order of their creation
		// if they are no-longer needed
			
			
			
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException sqlEx) { } // ignore
					rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException sqlEx) { } // ignore
					stmt = null;
			}
		}
		
		} catch (SQLException ex) {
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		}
		
		return data;
	}
	
	@Override
	public Instances parsePersons(String personFile) {
		// TODO Auto-generated method stub
		Instances       data = null;
		FastVector      atts;
		/**
		 * attributes list
		 */
		
		ArrayList attributelist=new ArrayList();
		/**
		 * id name location description  url
		 */
//		attributelist.add("ID");
//		attributelist.add("NAME");
//		attributelist.add("LOCATION");
//		attributelist.add("DESCRIPTION");
//		attributelist.add("URL");
//		
		Iterator attrit=this.personmappinghm.keySet().iterator();
		HashMap mappingattrhm=new HashMap();
		while(attrit.hasNext())
		{
			String tempattrstr=(String) attrit.next();
			attributelist.add(tempattrstr);
			MappingAttribute mappinga=(MappingAttribute) this.personmappinghm.get(tempattrstr);
			mappingattrhm.put(mappinga.getModelAttributeName(), mappinga);
		}
		
		/**
		 * connect to db :Mysql
		 */
		
		Connection conn = null;

		try {
		System.out.println("before connecting...");
		conn =DriverManager.getConnection(this.URL+
		"?user="+this.username+"&password="+this.password);
		System.out.println("connect successfully");
		// Do something with the Connection
		Statement stmt = null;
		ResultSet rs = null;
		try {
		stmt = (Statement) conn.createStatement();
		//rs = stmt.executeQuery("SELECT foo FROM bar");
		// or alternatively, if you don't know ahead of time that
		// the query will be a SELECT...
		
		 stmt.execute("set search_path to twitware,public;");
		 System.out.println("set search_path to twitware,public;");
		 stmt.setFetchSize(100); 
		 
			// 1. set up attributes
			
		    atts = new FastVector();
		    /**
			 * id name location description  url
			 * 
			 */
		   Set personmodelkeyset= this.personmodelhm.keySet();
		   List list=new ArrayList(personmodelkeyset);
		   Collections.sort(list);
		    Iterator it=list.iterator();
		    while(it.hasNext())
		    {
		    	String tempstr=(String)it.next();
//		    	System.out.println("tempstr1="+tempstr);
		    	ModelAttribute attribute=(ModelAttribute)this.personmodelhm.get(tempstr.trim());
		    	String attributename=attribute.getAttributeName();
//		    	System.out.println("attributename:"+attributename);
		    	int typeid=attribute.getAttributeTypeId();
		    	
		    	switch(typeid)
		    	{
		    	case 1:
		    		//-numeric
		    		
		    		atts.addElement(new Attribute(attributename));
		    		break;
		    	case 2:
		    		//-nominal
		    		FastVector attVals;
		    	    attVals = new FastVector();
		    	    ArrayList al=attribute.getNomialValues();
		    	    Iterator ital=al.iterator();
		    	    while(ital.hasNext())
		    	    {
		    	    	String alstr=(String) ital.next();
		    	    	attVals.addElement(alstr);
		    	    }
		    	    atts.addElement(new Attribute(attributename, attVals));
		    	    break;
		    		
		    	case 3:
		    		//-string
		    		atts.addElement(new Attribute(attributename, (FastVector) null));
		    		break;
		    	case 4:
		    		//-date
		    	    atts.addElement(new Attribute(attributename, (FastVector) null));
		    	    break;
		    	}
		    }
		 // - string
		    // instance_id and label
		    Attribute instanceid=new Attribute("INSTANCE_ID", (FastVector) null);
		    Attribute label=new Attribute("LABEL", (FastVector) null);
		    atts.addElement(instanceid);
		    atts.addElement(label);
		    
		    
		 // 2. create Instances object
		    data = new Instances("TwitterPersons", atts, 0);
		    
		    Instances tempdata=data;
		    HashMap hm=this.personMapping(attributelist);
		    
		   
		    FileWriter writer = null;
		    BufferedWriter bw=null;
	    	 
	    	 int offsetnum = 0;
			 int totalnum=0;
			 int limit=5000;
			 HashMap inhm=new HashMap();
		 while(true)
		 {
			   
			 //create file
			    String arfffilename=this.instancesdirectory+offsetnum+"_"+this.twitterpersonarfffilename;   
			    File arfffile=new File(arfffilename);
			    if(!arfffile.exists())
			    {
			    	try {
			    		arfffile.createNewFile();
					 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterpersonarfffilename);
		 
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    else
			    {
			    	try {
				    	if(this.isappended==1)
			    		{
							arfffile.delete();
							arfffile.createNewFile();
						 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterpersonarfffilename);
			    		}
			    	
		    		} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    
			// if()
			 stmt.execute("SELECT id,name,location,description,url FROM user_info limit "+limit+" offset "+offsetnum+";");
			
			// System.out.println("SELECT id,name,location,description,url FROM user_info limit 5000 offset "+offsetnum+";");
			 System.out.println("start query "); 
			rs = stmt.getResultSet();
			System.out.println("done query "); 
		 

		
		// Now do something with the ResultSet ....
		 // convert resultset to instances
			System.out.println("converting resultset to instances");
			
			if(rs!=null)
			{
		
			 // 3. fill with data
	
			    
			    
			    int num=0;
			    while(rs.next())
			    {
				    num++;
//			    	System.out.println("num0="+num);
			    	/**
					 * id name location description  url
					 */
			    	
			    	Iterator it1=attributelist.iterator();
					while(it1.hasNext())
					{
						String attstr=(String)it1.next();
	
						
						
						String rows = null;
						
						rows=rs.getString(attstr.toLowerCase());
//						System.out.println(rows);
						ModelAttribute attribute=(ModelAttribute)hm.get(attstr);
						String attributename=attribute.getAttributeName();
						inhm.put(attributename.trim(), rows);
					}
				    Instance newInst = new Instance(data.numAttributes());
					for(int i=0;i<data.numAttributes();i++)
					{
						
		//				System.out.println("i="+data.numAttributes());
						if(inhm.containsKey(data.attribute(i).name()))
						{
							if(inhm.get(data.attribute(i).name())==null)
								continue;
							
							if(data.attribute(i).name().equalsIgnoreCase("id"))
							{
								String key=(String)inhm.get(data.attribute(i).name());
								if(this.personspamhm.containsKey(key))
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "spam");
									
								}
								else
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "ham");
								}
							}
							
							
							
						 //-numeric
						   if(data.attribute(i).isNumeric())
						   {
							   newInst.setValue(data.attribute(i),Double.valueOf((String) inhm.get(data.attribute(i).name())));
							   
						   }
						   //-nominal
						   if(data.attribute(i).isNominal())
						   {
							   MappingAttribute mapa=(MappingAttribute) mappingattrhm.get(data.attribute(i).name());
							   ModelAttribute modela=(ModelAttribute) hm.get(mapa.getAttributeName());
							   String tempstr=(String) inhm.get(data.attribute(i).name());
							   ArrayList modelal=modela.getNomialValues();
							   ArrayList mappingal=mapa.getNomialValues();
							   HashMap mappingnominal=new HashMap();
							   for(int n=0;n<modelal.size();n++)
							   {
//								   System.out.println("mapping:"+mappingal.get(n)+ modelal.get(n));
								   mappingnominal.put(mappingal.get(n), modelal.get(n));
							   }
							   
							   newInst.setValue(data.attribute(i),(String)mappingnominal.get(tempstr));
						   }
						   //-string
						   if(data.attribute(i).isString())
						   {
		//					   System.out.println("--"+data.attribute(i).name());
							   String tempstr=(String)inhm.get(data.attribute(i).name());
	//						   System.out.println(tempstr);
							   
							   newInst.setValue(data.attribute(i),tempstr);
								
		//					   vals[i] = data.attribute(i).addStringValue("This is a string!");
						   }
					   
						   //-date
						   if(data.attribute(i).isDate())
						   {
							   newInst.setValue(data.attribute(i),(String) inhm.get(data.attribute(i).name()) );
								
		//					   try {
		//						vals[i] = data.attribute(i).parseDate("2001-11-09");
		//					} catch (ParseException e) {
		//						// TODO Auto-generated catch block
		//						e.printStackTrace();
		//					}
						   }
						}
					}
					
					int id=totalnum+num;
					String instanceidstr="Twitter_"+id;
					
					newInst.setValue(data.attribute(data.numAttributes()-2), instanceidstr);
					
					
				    // add
				    data.add(newInst);
				    newInst=null;
				
				    
				    
				    
			
			    }
			    System.out.println("num="+num);
			    totalnum=totalnum+num;
			    System.out.println("totalnum="+totalnum);
			    if(num==0)
			    	break;
			 // 4. output data
	//		    System.out.println(data);
			    
			   
			}
			else break;
			
		  
		   
		    // first time saving 
		 
		    	
		  
		   
		    	 //append new instance to the file
		    	
		    	writer =new FileWriter(arfffilename, true);
		    	
			    bw=new BufferedWriter(writer);
			    try {
			    	
			    	  // You are now at the end of the file,
			    	  // and can start writing new data out, e.g.
			    	 bw.newLine(); 
			    	 for(int i=0;i<data.numInstances();i++)
			    	 {
			    		 bw.write(data.instance(i).toString());
			    		 bw.newLine();
			    		 bw.flush();
			    	 }
			    	 bw.close();
			    	 writer.close();
			    	 bw=null;
			    	 writer=null;
			    	} catch (IOException ex ) {
			    	  ex.printStackTrace();
			    	}finally {                       // always close the file
			    		 if (bw != null) try {
			    		    bw.close();
			    		 } catch (IOException ioe2) {
			    		    // just ignore it
			    		 }
			    	 }
			    	// delete instances in data
			    data.delete();
		    
		    
//		    Runtime.getRuntime().gc();
			offsetnum=offsetnum+limit;
		 }
		}
		catch (SQLException ex){
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally {
		// it is a good idea to release
		// resources in a finally{} block
		// in reverse-order of their creation
		// if they are no-longer needed
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException sqlEx) { } // ignore
					rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException sqlEx) { } // ignore
					stmt = null;
			}
		}
		
		} catch (SQLException ex) {
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		}
		
		return data;
	}


	public Instances parseMessagesByStatusID(String msgFile) {
		// TODO Auto-generated method stub
		
		Instances       data = null;
		FastVector      atts;
		/**
		 * attributes list
		 */
		HashSet<String> statushs=this.loadFile2HashSet(msgFile);
		ArrayList attributelist=new ArrayList();
		/**
		 * id name location description  url
		 */
//		attributelist.add("ID");
//		attributelist.add("NAME");
//		attributelist.add("LOCATION");
//		attributelist.add("DESCRIPTION");
//		attributelist.add("URL");
//		
		Iterator attrit=this.messagemappinghm.keySet().iterator();
		HashMap mappingattrhm=new HashMap();
		while(attrit.hasNext())
		{
			String tempattrstr=(String) attrit.next();
			attributelist.add(tempattrstr);
			MappingAttribute mappinga=(MappingAttribute) this.messagemappinghm.get(tempattrstr);
			mappingattrhm.put(mappinga.getModelAttributeName(), mappinga);
		}
		
		/**
		 * connect to db :Mysql
		 */
		
		Connection conn = null;

		try {
		System.out.println("before connecting...");
		conn =DriverManager.getConnection("jdbc:postgresql://matcha.cc.gatech.edu:5432/twitware?" +
		"user=wang6&password=?WDcj984615?");
		System.out.println("connect successfully");
		// Do something with the Connection
		Statement stmt = null;
		ResultSet rs = null;
		try {
		stmt = (Statement) conn.createStatement();
		//rs = stmt.executeQuery("SELECT foo FROM bar");
		// or alternatively, if you don't know ahead of time that
		// the query will be a SELECT...
		
		 stmt.execute("set search_path to twitware,public;");
		 System.out.println("set search_path to twitware,public;");
		 stmt.setFetchSize(100); 
		 
			// 1. set up attributes
			
		    atts = new FastVector();
		    /**
			 * id name location description  url
			 * 
			 */
		   Set personmodelkeyset= this.messagemodelhm.keySet();
		   List list=new ArrayList(personmodelkeyset);
		   Collections.sort(list);
		    Iterator it=list.iterator();
		    while(it.hasNext())
		    {
		    	String tempstr=(String)it.next();
//		    	System.out.println("tempstr1="+tempstr);
		    	ModelAttribute attribute=(ModelAttribute)this.messagemodelhm.get(tempstr.trim());
		    	String attributename=attribute.getAttributeName();
//		    	System.out.println("attributename:"+attributename);
		    	int typeid=attribute.getAttributeTypeId();
		    	
		    	switch(typeid)
		    	{
		    	case 1:
		    		//-numeric
		    		
		    		atts.addElement(new Attribute(attributename));
		    		break;
		    	case 2:
		    		//-nominal
		    		FastVector attVals;
		    	    attVals = new FastVector();
		    	    ArrayList al=attribute.getNomialValues();
		    	    Iterator ital=al.iterator();
		    	    while(ital.hasNext())
		    	    {
		    	    	String alstr=(String) ital.next();
		    	    	attVals.addElement(alstr);
		    	    }
		    	    atts.addElement(new Attribute(attributename, attVals));
		    	    break;
		    		
		    	case 3:
		    		//-string
		    		atts.addElement(new Attribute(attributename, (FastVector) null));
		    		break;
		    	case 4:
		    		//-date
		    	    atts.addElement(new Attribute(attributename, (FastVector) null));
		    	    break;
		    	}
		    }
		 // - string
		    // instance_id and label
		    Attribute instanceid=new Attribute("INSTANCE_ID", (FastVector) null);
		    Attribute label=new Attribute("LABEL", (FastVector) null);
		    atts.addElement(instanceid);
		    atts.addElement(label);
		    
		    
		 // 2. create Instances object
		    data = new Instances("TwitterMessages", atts, 0);
		    Instances tempdata=data;
		    HashMap hm=this.messageMapping(attributelist);
		    
		   
		    FileWriter writer = null;
		    BufferedWriter bw=null;
	    	 
	    	 int offsetnum = 0;
			 int totalnum=0;
			 int limit=5000;
			 HashMap inhm=new HashMap();
		
			 
			 
			 //create file
			    String arfffilename=this.instancesdirectory+offsetnum+"_"+this.twittermessagearfffilename;   
			    File arfffile=new File(arfffilename);
			    if(!arfffile.exists())
			    {
			    	try {
			    		arfffile.createNewFile();
					 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twittermessagearfffilename);
		 
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    else
			    {
			    	try {
				    	if(this.isappended==1)
			    		{
							arfffile.delete();
							arfffile.createNewFile();
						 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterpersonarfffilename);
			    		}
			    	
		    		} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    
			 stmt.execute("SELECT id,user_id,text,in_reply_to_user_id FROM status ;");
			
			 System.out.println("SELECT id,user_id,text,in_reply_to_user_id FROM status limit 5000 offset "+offsetnum+";");
			 System.out.println("start query "); 
			rs = stmt.getResultSet();
			System.out.println("done query "); 
		 

		
		// Now do something with the ResultSet ....
		 // convert resultset to instances
			System.out.println("converting resultset to instances");
			
			if(rs!=null)
			{
		
			 // 3. fill with data
	
			    
			    
			    int num=0;
			    while(rs.next())
			    {
			    	String tempSID=rs.getString("id");
			    	if(!statushs.contains(tempSID.trim()))
			    		continue;
			    	
				    num++;
//			    	System.out.println("num0="+num);
			    	/**
					 * id name location description  url
					 */
			    	
			    	Iterator it1=attributelist.iterator();
					while(it1.hasNext())
					{
						String attstr=(String)it1.next();
						String rows = null;
						
						rows=rs.getString(attstr.toLowerCase());
//						System.out.println(rows);
						ModelAttribute attribute=(ModelAttribute)hm.get(attstr);
						String attributename=attribute.getAttributeName();
						inhm.put(attributename.trim(), rows);
					}
				    Instance newInst = new Instance(data.numAttributes());
					for(int i=0;i<data.numAttributes();i++)
					{
		//				System.out.println("i="+data.numAttributes());
						if(inhm.containsKey(data.attribute(i).name()))
						{
							if(inhm.get(data.attribute(i).name())==null)
								break;
							
							if(data.attribute(i).name().equalsIgnoreCase("id"))
							{
								String key=(String)inhm.get(data.attribute(i).name());
								if(this.messagespamhm.containsKey(key))
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "spam");
									
								}
								else
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "ham");
								}
							}
							
							
						 //-numeric
						   if(data.attribute(i).isNumeric())
						   {
							   newInst.setValue(data.attribute(i),Double.valueOf((String) inhm.get(data.attribute(i).name())));
							   
						   }
						   //-nominal
						   if(data.attribute(i).isNominal())
						   {
							   MappingAttribute mapa=(MappingAttribute) mappingattrhm.get(data.attribute(i).name());
							   ModelAttribute modela=(ModelAttribute) hm.get(mapa.getAttributeName());
							   String tempstr=(String) inhm.get(data.attribute(i).name());
							   ArrayList modelal=modela.getNomialValues();
							   ArrayList mappingal=mapa.getNomialValues();
							   HashMap mappingnominal=new HashMap();
							   for(int n=0;n<modelal.size();n++)
							   {
//								   System.out.println("mapping:"+mappingal.get(n)+ modelal.get(n));
								   mappingnominal.put(mappingal.get(n), modelal.get(n));
							   }
							   
							   newInst.setValue(data.attribute(i),(String)mappingnominal.get(tempstr));
						   }
						   //-string
						   if(data.attribute(i).isString())
						   {
		//					   System.out.println("--"+data.attribute(i).name());
							   String tempstr=(String)inhm.get(data.attribute(i).name());
	//						   System.out.println(tempstr);
							   
							   newInst.setValue(data.attribute(i),tempstr);
								
		//					   vals[i] = data.attribute(i).addStringValue("This is a string!");
						   }
					   
						   //-date
						   if(data.attribute(i).isDate())
						   {
							   newInst.setValue(data.attribute(i),(String) inhm.get(data.attribute(i).name()) );
								
		//					   try {
		//						vals[i] = data.attribute(i).parseDate("2001-11-09");
		//					} catch (ParseException e) {
		//						// TODO Auto-generated catch block
		//						e.printStackTrace();
		//					}
						   }
						}
					}
					
					int id=totalnum+num;
					String instanceidstr="Twitter_"+id;
					
					newInst.setValue(data.attribute(data.numAttributes()-2), instanceidstr);
					
				    // add
				    data.add(newInst);
				    newInst=null;
				
				    
				    
				    
			
			    }
			    System.out.println("num="+num);
			    totalnum=totalnum+num;
			    System.out.println("totalnum="+totalnum);
			   
			 // 4. output data
	//		    System.out.println(data);
			    
			   
			}
		
			
		  
		   
		    // first time saving 
		 
		    	
		  
		   
		    	 //append new instance to the file
		    	
		    	writer =new FileWriter(arfffilename, true);
		    	
			    bw=new BufferedWriter(writer);
			    try {
			    	
			    	  // You are now at the end of the file,
			    	  // and can start writing new data out, e.g.
			    	 bw.newLine(); 
			    	 for(int i=0;i<data.numInstances();i++)
			    	 {
			    		 bw.write(data.instance(i).toString());
			    		 bw.newLine();
			    		 bw.flush();
			    	 }
			    	 bw.close();
			    	 writer.close();
			    	 bw=null;
			    	 writer=null;
			    	} catch (IOException ex ) {
			    	  ex.printStackTrace();
			    	}finally {                       // always close the file
			    		 if (bw != null) try {
			    		    bw.close();
			    		 } catch (IOException ioe2) {
			    		    // just ignore it
			    		 }
			    	 }
			    	// delete instances in data
			    data.delete();
		    
		    
//		    Runtime.getRuntime().gc();
			offsetnum=offsetnum+limit;
		
		}
		catch (SQLException ex){
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally {
		// it is a good idea to release
		// resources in a finally{} block
		// in reverse-order of their creation
		// if they are no-longer needed
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException sqlEx) { } // ignore
					rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException sqlEx) { } // ignore
					stmt = null;
			}
		}
		
		} catch (SQLException ex) {
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		}
		
		return data;
	}

	public Instances parseMessages(String msgFile) {
		// TODO Auto-generated method stub
		
		Instances       data = null;
		FastVector      atts;
		/**
		 * attributes list
		 */
		
		ArrayList attributelist=new ArrayList();
		/**
		 * id name location description  url
		 */
//		attributelist.add("ID");
//		attributelist.add("NAME");
//		attributelist.add("LOCATION");
//		attributelist.add("DESCRIPTION");
//		attributelist.add("URL");
//		
		Iterator attrit=this.messagemappinghm.keySet().iterator();
		HashMap mappingattrhm=new HashMap();
		while(attrit.hasNext())
		{
			String tempattrstr=(String) attrit.next();
			attributelist.add(tempattrstr);
			MappingAttribute mappinga=(MappingAttribute) this.messagemappinghm.get(tempattrstr);
			mappingattrhm.put(mappinga.getModelAttributeName(), mappinga);
		}
		
		/**
		 * connect to db :Mysql
		 */
		
		Connection conn = null;

		try {
		System.out.println("before connecting...");
		conn =DriverManager.getConnection("jdbc:postgresql://matcha.cc.gatech.edu:5432/twitware?" +
		"user=wang6&password=?WDcj984615?");
		System.out.println("connect successfully");
		// Do something with the Connection
		Statement stmt = null;
		ResultSet rs = null;
		try {
		stmt = (Statement) conn.createStatement();
		//rs = stmt.executeQuery("SELECT foo FROM bar");
		// or alternatively, if you don't know ahead of time that
		// the query will be a SELECT...
		
		 stmt.execute("set search_path to twitware,public;");
		 System.out.println("set search_path to twitware,public;");
		 stmt.setFetchSize(100); 
		 
			// 1. set up attributes
			
		    atts = new FastVector();
		    /**
			 * id name location description  url
			 * 
			 */
		   Set personmodelkeyset= this.messagemodelhm.keySet();
		   List list=new ArrayList(personmodelkeyset);
		   Collections.sort(list);
		    Iterator it=list.iterator();
		    while(it.hasNext())
		    {
		    	String tempstr=(String)it.next();
//		    	System.out.println("tempstr1="+tempstr);
		    	ModelAttribute attribute=(ModelAttribute)this.messagemodelhm.get(tempstr.trim());
		    	String attributename=attribute.getAttributeName();
//		    	System.out.println("attributename:"+attributename);
		    	int typeid=attribute.getAttributeTypeId();
		    	
		    	switch(typeid)
		    	{
		    	case 1:
		    		//-numeric
		    		
		    		atts.addElement(new Attribute(attributename));
		    		break;
		    	case 2:
		    		//-nominal
		    		FastVector attVals;
		    	    attVals = new FastVector();
		    	    ArrayList al=attribute.getNomialValues();
		    	    Iterator ital=al.iterator();
		    	    while(ital.hasNext())
		    	    {
		    	    	String alstr=(String) ital.next();
		    	    	attVals.addElement(alstr);
		    	    }
		    	    atts.addElement(new Attribute(attributename, attVals));
		    	    break;
		    		
		    	case 3:
		    		//-string
		    		atts.addElement(new Attribute(attributename, (FastVector) null));
		    		break;
		    	case 4:
		    		//-date
		    	    atts.addElement(new Attribute(attributename, (FastVector) null));
		    	    break;
		    	}
		    }
		 // - string
		    // instance_id and label
		    Attribute instanceid=new Attribute("INSTANCE_ID", (FastVector) null);
		    Attribute label=new Attribute("LABEL", (FastVector) null);
		    atts.addElement(instanceid);
		    atts.addElement(label);
		    
		    
		 // 2. create Instances object
		    data = new Instances("TwitterMessages", atts, 0);
		    Instances tempdata=data;
		    HashMap hm=this.messageMapping(attributelist);
		    
		   
		    FileWriter writer = null;
		    BufferedWriter bw=null;
	    	 
	    	 int offsetnum = 0;
			 int totalnum=0;
			 int limit=5000;
			 HashMap inhm=new HashMap();
		 while(true)
		 {
			 
			 
			 //create file
			    String arfffilename=this.instancesdirectory+offsetnum+"_"+this.twittermessagearfffilename;   
			    File arfffile=new File(arfffilename);
			    if(!arfffile.exists())
			    {
			    	try {
			    		arfffile.createNewFile();
					 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twittermessagearfffilename);
		 
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    else
			    {
			    	try {
				    	if(this.isappended==1)
			    		{
							arfffile.delete();
							arfffile.createNewFile();
						 	this.save2arff(tempdata, this.instancesdirectory, offsetnum+"_"+this.twitterpersonarfffilename);
			    		}
			    	
		    		} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    
			 stmt.execute("SELECT id,user_id,text,in_reply_to_user_id FROM status limit "+limit+" offset "+offsetnum+";");
			
			 System.out.println("SELECT id,user_id,text,in_reply_to_user_id FROM status limit 5000 offset "+offsetnum+";");
			 System.out.println("start query "); 
			rs = stmt.getResultSet();
			System.out.println("done query "); 
		 

		
		// Now do something with the ResultSet ....
		 // convert resultset to instances
			System.out.println("converting resultset to instances");
			
			if(rs!=null)
			{
		
			 // 3. fill with data
	
			    
			    
			    int num=0;
			    while(rs.next())
			    {
				    num++;
//			    	System.out.println("num0="+num);
			    	/**
					 * id name location description  url
					 */
			    	
			    	Iterator it1=attributelist.iterator();
					while(it1.hasNext())
					{
						String attstr=(String)it1.next();
						String rows = null;
						
						rows=rs.getString(attstr.toLowerCase());
//						System.out.println(rows);
						ModelAttribute attribute=(ModelAttribute)hm.get(attstr);
						String attributename=attribute.getAttributeName();
						inhm.put(attributename.trim(), rows);
					}
				    Instance newInst = new Instance(data.numAttributes());
					for(int i=0;i<data.numAttributes();i++)
					{
		//				System.out.println("i="+data.numAttributes());
						if(inhm.containsKey(data.attribute(i).name()))
						{
							if(inhm.get(data.attribute(i).name())==null)
								break;
							
							if(data.attribute(i).name().equalsIgnoreCase("id"))
							{
								String key=(String)inhm.get(data.attribute(i).name());
								if(this.messagespamhm.containsKey(key))
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "spam");
									
								}
								else
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "ham");
								}
							}
							
							
						 //-numeric
						   if(data.attribute(i).isNumeric())
						   {
							   newInst.setValue(data.attribute(i),Double.valueOf((String) inhm.get(data.attribute(i).name())));
							   
						   }
						   //-nominal
						   if(data.attribute(i).isNominal())
						   {
							   MappingAttribute mapa=(MappingAttribute) mappingattrhm.get(data.attribute(i).name());
							   ModelAttribute modela=(ModelAttribute) hm.get(mapa.getAttributeName());
							   String tempstr=(String) inhm.get(data.attribute(i).name());
							   ArrayList modelal=modela.getNomialValues();
							   ArrayList mappingal=mapa.getNomialValues();
							   HashMap mappingnominal=new HashMap();
							   for(int n=0;n<modelal.size();n++)
							   {
//								   System.out.println("mapping:"+mappingal.get(n)+ modelal.get(n));
								   mappingnominal.put(mappingal.get(n), modelal.get(n));
							   }
							   
							   newInst.setValue(data.attribute(i),(String)mappingnominal.get(tempstr));
						   }
						   //-string
						   if(data.attribute(i).isString())
						   {
		//					   System.out.println("--"+data.attribute(i).name());
							   String tempstr=(String)inhm.get(data.attribute(i).name());
	//						   System.out.println(tempstr);
							   
							   newInst.setValue(data.attribute(i),tempstr);
								
		//					   vals[i] = data.attribute(i).addStringValue("This is a string!");
						   }
					   
						   //-date
						   if(data.attribute(i).isDate())
						   {
							   newInst.setValue(data.attribute(i),(String) inhm.get(data.attribute(i).name()) );
								
		//					   try {
		//						vals[i] = data.attribute(i).parseDate("2001-11-09");
		//					} catch (ParseException e) {
		//						// TODO Auto-generated catch block
		//						e.printStackTrace();
		//					}
						   }
						}
					}
					
					int id=totalnum+num;
					String instanceidstr="Twitter_"+id;
					
					newInst.setValue(data.attribute(data.numAttributes()-2), instanceidstr);
					
				    // add
				    data.add(newInst);
				    newInst=null;
				
				    
				    
				    
			
			    }
			    System.out.println("num="+num);
			    totalnum=totalnum+num;
			    System.out.println("totalnum="+totalnum);
			    if(num==0)
			    	break;
			 // 4. output data
	//		    System.out.println(data);
			    
			   
			}
			else break;
			
		  
		   
		    // first time saving 
		 
		    	
		  
		   
		    	 //append new instance to the file
		    	
		    	writer =new FileWriter(arfffilename, true);
		    	
			    bw=new BufferedWriter(writer);
			    try {
			    	
			    	  // You are now at the end of the file,
			    	  // and can start writing new data out, e.g.
			    	 bw.newLine(); 
			    	 for(int i=0;i<data.numInstances();i++)
			    	 {
			    		 bw.write(data.instance(i).toString());
			    		 bw.newLine();
			    		 bw.flush();
			    	 }
			    	 bw.close();
			    	 writer.close();
			    	 bw=null;
			    	 writer=null;
			    	} catch (IOException ex ) {
			    	  ex.printStackTrace();
			    	}finally {                       // always close the file
			    		 if (bw != null) try {
			    		    bw.close();
			    		 } catch (IOException ioe2) {
			    		    // just ignore it
			    		 }
			    	 }
			    	// delete instances in data
			    data.delete();
		    
		    
//		    Runtime.getRuntime().gc();
			offsetnum=offsetnum+limit;
		 }
		}
		catch (SQLException ex){
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally {
		// it is a good idea to release
		// resources in a finally{} block
		// in reverse-order of their creation
		// if they are no-longer needed
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException sqlEx) { } // ignore
					rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException sqlEx) { } // ignore
					stmt = null;
			}
		}
		
		} catch (SQLException ex) {
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		}
		
		return data;
	}

	
/**
 * attributes mapping
 * @param attr
 * @return
 */
	public HashMap personMapping(ArrayList attr)
	{
		
		HashMap hm=new HashMap();
		Iterator it=attr.iterator();
		while(it.hasNext())
		{
			String tempstr=(String) it.next();
//			System.out.println("personMapping tempstr="+tempstr);
			/**
			 * id name location description  url
			 */
			MappingAttribute maa=(MappingAttribute)(this.personmappinghm.get(tempstr));
			String modelattributename=maa.getModelAttributeName();
//			System.out.println("modelattributename:"+modelattributename);
			ModelAttribute ma=(ModelAttribute)this.personmodelhm.get(modelattributename);

//			System.out.println("ma.getattributename:"+ma.getAttributeName());
			hm.put(tempstr,ma);
		}
		return hm;
		
	}
	
	
	/**
	 * attributes mapping
	 * @return
	 */
	public HashMap messageMapping(ArrayList attr)
	{

		HashMap hm=new HashMap();
		Iterator it=attr.iterator();
		while(it.hasNext())
		{
			String tempstr=(String) it.next();
//			System.out.println("personMapping tempstr="+tempstr);
			/**
			 * id, user_id, text, in_reply_to_user_id
			 */
			MappingAttribute maa=(MappingAttribute)(this.messagemappinghm.get(tempstr));
			String modelattributename=maa.getModelAttributeName();
//			System.out.println("modelattributename:"+modelattributename);
			ModelAttribute ma=(ModelAttribute)this.messagemodelhm.get(modelattributename);

//			System.out.println("ma.getattributename:"+ma.getAttributeName());
			hm.put(tempstr,ma);
		}
		return hm;
		


	}
	/**
	 * attributes mapping
	 * @return
	 */
	public HashMap webpageMapping(ArrayList attr)
	{

		HashMap hm=new HashMap();
		Iterator it=attr.iterator();
		while(it.hasNext())
		{
			String tempstr=(String) it.next();
//			System.out.println("personMapping tempstr="+tempstr);
			/**
			 * id, user_id, text, in_reply_to_user_id
			 */
			MappingAttribute maa=(MappingAttribute)(this.webpagemappinghm.get(tempstr));
			String modelattributename=maa.getModelAttributeName();
//			System.out.println("modelattributename:"+modelattributename);
			ModelAttribute ma=(ModelAttribute)this.webpagemodelhm.get(modelattributename);

//			System.out.println("ma.getattributename:"+ma.getAttributeName());
			hm.put(tempstr,ma);
		}
		return hm;
	}


	
	/**
	 * save instances to arff files
	 * @param in
	 * @param filedirectory
	 * @param filename
	 */
	public void save2arff(Instances in, String filedirectory, String filename)
	{
		 Instances dataSet =in;
		 ArffSaver saver = new ArffSaver();
		 saver.setInstances(dataSet);
		 try {
			saver.setFile(new File(filedirectory+System.getProperties().getProperty("file.separator")+filename));
		//	saver.setDestination(new File("./data/test.arff"));   // **not** necessary in 3.5.4 and later
			 saver.writeBatch();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		 
	}
	
	
	
	public int getIsappended() {
		return isappended;
	}


	public void setIsappended(int isappended) {
		this.isappended = isappended;
	}


	public static void  main(String [] arg0)
	{
		
		TwitterMapper tm=new TwitterMapper();
		tm.init();
	
		Instances data=tm.parseMessages(null);
//		tm.save2arff(data, System.getProperties().getProperty("user.dir"), "test.arff");
	}
}
